/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

import org.json.simple.JSONObject;
import org.sss.client.DataGetter;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.datatype.PackableJSONObject;
import org.sss.mapreduce.datatype.PackableString;

public class ContApacheLogGetData {
  private final SssClient client;
  public ContApacheLogGetData(SssClient client) throws SssException {
    this.client = client;
  }
  private static void usage() {
    System.exit(1);
  }
  private <T> Iterator<T> sortedIterator(Iterator<T> it, Comparator<T> comparator) {
    List<T> list = new ArrayList<T>();
    while (it.hasNext()) {
      list.add(it.next());
    }
    Collections.sort(list, comparator);
    return list.iterator();
  }
  private Comparator<Tuple<PackableString, PackableJSONObject>> keyValueComparator() {
    return new Comparator<Tuple<PackableString, PackableJSONObject>>() {
      @Override
      public int compare(Tuple<PackableString, PackableJSONObject> o1,
                         Tuple<PackableString, PackableJSONObject> o2) {
        String key1 = o1.getKey().get();
        String key2 = o2.getKey().get();
        int keyCmp = key1.compareTo(key2);
        if (keyCmp == 0) {
          JSONObject json1 = o1.getValue().get();
          JSONObject json2 = o2.getValue().get();
          int scheduled1 = ((Number) json1.get("scheduled")).intValue();
          int scheduled2 = ((Number) json2.get("scheduled")).intValue();
          if (scheduled1 == scheduled2) {
            int sum1 = ((Number) json1.get("sum")).intValue();
            int sum2 = ((Number) json2.get("sum")).intValue();
            if (sum1 == sum2) {
              return 0;
            }
            else {
              return (sum1 < sum2) ? -1 : 1;
            }
          }
          else {
            return (scheduled1 < scheduled2) ? -1 : 1;
          }
        }
        else {
          return keyCmp;
        }
      }
      @Override
      public boolean equals(Object o) {
        return (this == o);
      }
    };
  }
  private void getData(GroupID gid) throws Exception {
    DataGetter<PackableString, PackableJSONObject> dg =
      DataGetter.create(client, PackableString.class, PackableJSONObject.class, gid);
    Iterator<Tuple<PackableString, PackableJSONObject>> it =
      sortedIterator(dg.iterator(), keyValueComparator());
    while (it.hasNext()) {
      Tuple<PackableString, PackableJSONObject> kv = it.next();
      String key = kv.getKey().get();
      JSONObject json = kv.getValue().get();
      int scheduled = ((Number) json.get("scheduled")).intValue();
      int sum = ((Number) json.get("sum")).intValue();
      System.out.println
        (String.format("key: %s, sum: %5d, scheduled: %4d",
                       key, sum, scheduled));
    }
  }
  /**
   * Usage:
   * java ContApacheLogGetData gid
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);
    String[] others = client.getArgs();
    if (others.length != 1) {
      usage();
    }
    GroupID gid = GroupID.createFromString(others[0]);
    ContApacheLogGetData app = new ContApacheLogGetData(client);
    app.getData(gid);
  }
}
